package com.wuba.rxjavademo.lyz.day0519;

import com.wuba.rxjavademo.lyz.BaseDemo;

import java.util.List;

import rx.Observable;
import rx.Subscriber;
import rx.functions.Action0;
import rx.schedulers.Schedulers;
import rx.subjects.PublishSubject;

/**
 * Created by liyaxing on 2016/5/20.
 */
public class Demo03 extends BaseDemo {

//    private static Exception sign = new Exception() ;

    public static void main(String[] ss){

        final Observable<Observable<Integer>> observableObservable = Observable.create(subscriber -> {
            for (int i = 0; i < 20; i++) {
                subscriber.onNext(createObservable(i));
            }
            subscriber.onCompleted();
        }) ;
        final Observable<Observable<Integer>> observableObservable2 = Observable.create(subscriber -> {
            for (int i = 0; i < 20; i++) {
                subscriber.onNext(createObservable2(i));
            }
            subscriber.onCompleted();
        }) ;
        Observable
                .mergeDelayError(observableObservable)
                .doOnSubscribe(new Action0() {
                    @Override
                    public void call() {
                        print("threadid : " + Thread.currentThread().getId());
                    }
                })
                .onErrorResumeNext(throwable -> {
                    print("throwable = " + throwable);
                    final PublishSubject<Integer> subject = PublishSubject.create() ;
                    Observable
                            .mergeDelayError(observableObservable2)
                            .subscribe(subject) ;
//                    Observable.create((Observable.OnSubscribe<Integer>) subscriber -> {
//                        try {
//                            Thread.sleep(1000);
//                        } catch (InterruptedException e) {
//                            e.printStackTrace();
//                        }
//                        subscriber.onNext(9999);
//                        subscriber.onCompleted();
//                    }).subscribeOn(Schedulers.io()).subscribe(subject);
                    return subject ;
                })
                .toList()
                .subscribe(new Subscriber<List<Integer>>() {
                    @Override
                    public void onCompleted() {
                        print("onCompleted");
                    }

                    @Override
                    public void onError(Throwable e) {
                        print("onError e : " + e );
                    }

                    @Override
                    public void onNext(List<Integer> integer) {
                        print("onNext integer : " + integer );
                    }
                }) ;

        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }

    public static Observable<Integer> createObservable(final int index){
        return Observable.create(new Observable.OnSubscribe<Integer>() {
            @Override
            public void call(Subscriber<? super Integer> subscriber) {
                if(index % 3 == 0){
                    subscriber.onError(new NullPointerException("xxx"));
                }else {
                    subscriber.onNext(index);
                    subscriber.onCompleted();
                }
            }
        }) ;
    }

    public static Observable<Integer> createObservable2(final int index){
        return Observable.create(new Observable.OnSubscribe<Integer>() {
            @Override
            public void call(Subscriber<? super Integer> subscriber) {
                try {
                    Thread.sleep(200);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                subscriber.onNext(index);
                subscriber.onCompleted();
            }
        }).subscribeOn(Schedulers.io()) ;
    }

}
